// License Notification
//======================================================
// Copyright (c) 2010 Channel Intelligence Inc.
// All rights reserved. This program and the accompanying materials
// are made available under the terms of the Eclipse Public License v1.0
// which accompanies this distribution, and is available at
// http://www.eclipse.org/legal/epl-v10.html
//
// ======================================================




using System;
using System.Collections.Generic;
using System.Linq;
using System.Xml.Linq;
using System.Text;
using CIP4.Common;
using CIP4.Simple;

using Aquiles;
using Aquiles.Command;
using Aquiles.Exceptions;
using Aquiles.Command.System;
using Aquiles.Model;

namespace CIP4.DataProvider.Cassandra
{
    public class DataProvider : Simple.DataProvider, IDataProvider
    {
        readonly string _keySpace;

        readonly string _columnFamily;

        const string ClusterName = "CIP4Cluster";

        public Encoding Utf8Encoding = Encoding.UTF8;


        #region IDataProvider Members

        public string ConnectionString { get; private set; }

        IDataProvider Cache { get; set; }

        public DataProvider(CIP4ConnectionAttribute DataProviderConnection, IDataProvider cachingProvider, LoggerDelegate logger)
            : base(DataProviderConnection)
        {
            if (!(DataProviderConnection is CIP4CassandraConnectionAttribute))
                LoggerWrapper(MajorEnum.Error, MinorEnum.UnexpectedConnectionType,
                    "{0} is not an acceptable connection type", new object[] { DataProviderConnection.ToString() }, logger);
            else
            {
                LoggerWrapper(MajorEnum.Trace, MinorEnum.InitializingProvider,
                    "Initalizing Provider: {0}", new object[] { DataProviderConnection }, logger);
                CIP4CassandraConnectionAttribute con = (CIP4CassandraConnectionAttribute)DataProviderConnection;
                _keySpace = con.Keyspace;
                _columnFamily = con.ColumnFamily;
                ConnectionString = DataProviderConnection.ToString();

                MakeColumnFamily(logger);

                Cache = cachingProvider;
            }
        }

        public void DeleteCollection(CollectionType collection, LoggerDelegate logger)
        {
            // Cassandra gets her knickers in a twist if more than one thread gets into the following code at the same time
            using (new CollectionLockType(collection.Name))
            {
                if (!LoggerWrapper(MajorEnum.Trace, MinorEnum.ActionStart,
                    "Started {0}", new object[] { MinorEnum.DeleteCollection }, logger))
                    return;

                DropColumnFamilyCommand cmd = new DropColumnFamilyCommand
                {
                    ColumnFamily = _columnFamily,
                    ConsistencyLevel = AquilesConsistencyLevel.ONE
                };
                try
                {
                    if (Cache != null)
                        Cache.DeleteCollection(collection, logger);

                    using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                    {
                        connection.Execute(cmd);
                        MakeColumnFamily(logger);
                    }
                    LoggerWrapper(MajorEnum.Information, MinorEnum.DeleteCollection,
                        "Collection deleted {0}", new object[] { collection }, logger);
                }
                catch (AquilesCommandParameterException ex)
                {
                    LoggerWrapper(MajorEnum.Error, MinorEnum.ErrorDeletingCollection,
                        "Error deleting collection: {0}", new object[] { ex }, logger);
                }
            }
        }

        public override IEnumerable<object> GetItems(CollectionType collection, IBooleanValue selectionExpression, CursorType cursor)
        {
            IList<object> items = new List<object>();
            if (!LoggerWrapper(MajorEnum.Trace, MinorEnum.ActionStart,
                "Started {0}", new object[] { MinorEnum.GetItems }, cursor.Logger))
                return items;

            try
            {
                ISerializerType serializer = cursor.GetSerializer(SerializerKindEnum.CIP4Xml);

                string lookForExternalId = GetExternalIdValueFromSelectionExpression(selectionExpression, serializer);

                if (!string.IsNullOrEmpty(lookForExternalId))
                {
                    // if we can find it in the cache, no point in looking in Cassandra, as an id has 
                    // been specified on the given template item. 
                    if (Cache != null)
                    {
                        foreach (object item in Cache.GetItems(collection, selectionExpression, cursor))
                        {
                            items.Add(item);
                            return items;
                        }
                    }

                    if (!string.IsNullOrEmpty(lookForExternalId))
                    {
                        GetCommand cmd = new GetCommand
                        {
                            ColumnFamily = _columnFamily,
                            Key = Utf8Encoding.GetBytes(lookForExternalId),
                            ColumnName = Utf8Encoding.GetBytes("Data")
                        };
                        try
                        {
                            using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                            {
                                connection.Execute(cmd);
                                if (cmd.Output != null && cmd.Output.Column != null)
                                {
                                    string itemText = Utf8Encoding.GetString(cmd.Output.Column.Value);
                                    object item = serializer.GetObjectForData(XElement.Parse(itemText));
                                    if (CompareProperties(selectionExpression, item, serializer))
                                    {
                                        items.Add(item);
                                        if (Cache != null)
                                            Cache.SaveItems(items, collection, cursor.GetSerializer);
                                    }
                                }
                            }
                        }
                        catch (AquilesCommandParameterException ex)
                        {
                            LoggerWrapper(MajorEnum.Error, MinorEnum.ErrorGettingItemsFromProvider,
                                "Error getting item with externalID={0}: {1}", new object[] { lookForExternalId, ex }, cursor.Logger);
                        }
                    }
                }
                else
                {
                    if (cursor.FailOnLinearSearch)
                        if (!LoggerWrapper(MajorEnum.Error, MinorEnum.FailOnLinearSearch,
                            "Failed on linear search accessing {0}", new object[] { collection }, cursor.Logger))
                            return items;

                    Encoding encoding = Encoding.UTF8;

                    int i = 0;

                    while (true)
                    {
                        Byte[] lastStart = string.IsNullOrEmpty(cursor.Marker) ?
                            new Byte[] { } :
                            encoding.GetBytes(cursor.Marker);

                        GetKeyRangeSliceCommand cmd = new GetKeyRangeSliceCommand
                        {
                            ColumnFamily = _columnFamily,
                            KeyTokenRange = new AquilesKeyRange
                            {
                                StartKey = lastStart,
                                EndKey = new Byte[] { },
                                Count = cursor.PageSize + 1
                            },
                            Predicate = new AquilesSlicePredicate
                            {
                                Columns = new List<byte[]> { encoding.GetBytes("Data") }
                            }
                        };

                        using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                        {
                            connection.Execute(cmd);
                            if (cmd.Output.Count == 0 || (cmd.Output.Count == 1 && encoding.GetString(cmd.Output[0].Key) == encoding.GetString(lastStart)))
                                break;
                            foreach (GetKeyRangeSliceCommand.Out result in cmd.Output)
                            {
                                if (encoding.GetString(result.Key) == encoding.GetString(lastStart))
                                    continue;
                                
                                cursor.Marker = encoding.GetString(result.Key);

                                foreach (GetCommand.Out result2 in result.Columns)
                                {
                                    string itemText = encoding.GetString(result2.Column.Value);
                                    object item = serializer.GetObjectForData(XElement.Parse(itemText));
                                    if (CompareProperties(selectionExpression, item, serializer))
                                    {
                                        items.Add(item);

                                        if (Cache != null)
                                            Cache.SaveItems(items, collection, cursor.GetSerializer);

                                        i += 1;
                                        if (i >= cursor.PageSize)
                                            return items;
                                    }
                                }
                            }
                        }
                    }
                }                    
                return items;
            }
            finally
            {
                LoggerWrapper(MajorEnum.Information, MinorEnum.GetItems,
                    "Get {0} items from {1}", new object[] { items.Count, collection }, cursor.Logger);
            }
        }

        public override void PurgeItems(IEnumerable<ExternalIDType> externalIDs, CollectionType collection, GetSerializerDelegate getSerializer)
        {
            ISerializerType serializer = getSerializer(SerializerKind);
            if (!LoggerWrapper(MajorEnum.Trace, MinorEnum.ActionStart,
                "Started {0}", new object[] { MinorEnum.PurgeItems }, serializer.Logger))
                return;

            int itemCount = 0;
            foreach (ExternalIDType externalID in externalIDs)
            {
                itemCount++;
                DeleteCommand cmd = 
                    new DeleteCommand
                    {
                        ColumnFamily = _columnFamily,
                        Key = Utf8Encoding.GetBytes(externalID.ToString())
                    };

                try
                {

                    using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                    {
                        connection.Execute(cmd);
                    }
                    LoggerWrapper(MajorEnum.Trace, MinorEnum.ItemDeleted, "deleted item {0}", new object[] { externalID }, serializer.Logger);
                }
                catch (AquilesCommandParameterException ex)
                {
                    LoggerWrapper(MajorEnum.Error, MinorEnum.ErrorPurgingItems,
                        "Error deleting item with externalID={0}: {1}", new object[] { externalID, ex }, serializer.Logger);
                }

            }

            if (Cache != null)
                Cache.PurgeItems(externalIDs, collection, getSerializer);

            LoggerWrapper(MajorEnum.Information, MinorEnum.PurgeItems,
                "{0} items deleted from {1}", new object[] { itemCount, collection.Name }, serializer.Logger);
        }

        public void SaveItems(IEnumerable<object> items, CollectionType collection, GetSerializerDelegate getSerializer)
        {
            ISerializerType serializer = getSerializer(SerializerKind);

            if (!LoggerWrapper(MajorEnum.Trace, MinorEnum.ActionStart,
                "Started {0}", new object[] { MinorEnum.SaveItems }, serializer.Logger))
                return;

            int itemCount = 0;
            foreach (object objectItem in items)
            {
                if (!SetHeaderProperties(objectItem, collection, serializer))
                    continue;

                InsertCommand cmd = new InsertCommand
                {
                    ColumnFamily = _columnFamily,
                    Key = Utf8Encoding.GetBytes(serializer.GetExternalID(objectItem).ToString()),
                    Column = new AquilesColumn
                    {
                        ColumnName = Utf8Encoding.GetBytes("Data"),
                        Value = Utf8Encoding.GetBytes(serializer.GetDataForObject(objectItem).ToString()),
                        //TTL = 852282524
                    }
                };

                try
                {
                    using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                    {
                        connection.Execute(cmd);
                        itemCount++;
                    }
                }
                catch (AquilesCommandParameterException ex)
                {
                    LoggerWrapper(MajorEnum.Error, MinorEnum.ErrorDeletingCollection,
                        "Error deleting collection: {0}", new object[] { ex }, serializer.Logger);
                }
                catch (Exception ex)
                {
                    LoggerWrapper(MajorEnum.Error, MinorEnum.ErrorDeletingCollection,
                        "Error deleting collection: {0}", new object[] { ex }, serializer.Logger);
                }

            }

            if (Cache != null)
                Cache.SaveItems(items, collection, getSerializer);

            LoggerWrapper(MajorEnum.Information, MinorEnum.SaveItems,
                "{0} items saved in {1}", new object[] { itemCount, collection.Name }, serializer.Logger);
        }

        public void Terminate(LoggerDelegate logger)
        {
        }

        #endregion

        #region Helper Functions

        private AquilesKeyspace MakeKeySpace(LoggerDelegate logger)
        {
            try
            {
                RetrieveKeyspacesCommand retrieveKeyspacesCmd = new RetrieveKeyspacesCommand();
                using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName))
                {
                    connection.Execute(retrieveKeyspacesCmd);
                }
                AquilesKeyspace keySpace = retrieveKeyspacesCmd.Keyspaces.FirstOrDefault(ks => ks.Name.CompareTo(_keySpace) == 0);
                if (keySpace != default(AquilesKeyspace))
                {
                    return keySpace;
                }

                AddKeyspaceCommand addKeyspaceCmd = 
                    new AddKeyspaceCommand
                    {
                        KeyspaceDefinition = 
                            new AquilesKeyspace
                                {
                                Name = _keySpace,
                                ReplicationFactor = 1,
                                ReplicationPlacementStrategy =
                                    AquilesKeyspace.
                                    SIMPLESTRATEGY
                            }
                    };

                using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName))
                {
                    connection.Execute(addKeyspaceCmd);
                }

                using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName))
                {
                    connection.Execute(retrieveKeyspacesCmd);
                }

                keySpace = retrieveKeyspacesCmd.Keyspaces.FirstOrDefault(ks => ks.Name.CompareTo(_keySpace) == 0);
                if (keySpace != default(AquilesKeyspace))
                {
                    LoggerWrapper(MajorEnum.Trace, MinorEnum.KeySpaceCreated,
                        "Created KeySpace {0}", new object[] { _keySpace }, logger);
                    return keySpace;
                }
                LoggerWrapper(MajorEnum.Error, MinorEnum.UnableToAcessKeySpaceAfterCreate,
                              "Unable to access KeySpace {0} after create", new object[] { _keySpace }, logger);
            }
            catch (Exception ex)
            {
                LoggerWrapper(MajorEnum.Error, MinorEnum.ExceptionCreatingKeySpace,
                    "Exception creating KeySpace {0}: {1}", new object[] { _keySpace, ex }, logger);
            }
            return null;
        }

        private void MakeColumnFamily(LoggerDelegate logger)
        {
            try
            {
                AquilesKeyspace keySpace = MakeKeySpace(logger);
                if (keySpace == default(AquilesKeyspace))
                    return;

                if (keySpace.ColumnFamilies.ContainsKey(_columnFamily))
                    return;

                AddColumnFamilyCommand cmd = new AddColumnFamilyCommand
                {
                    ColumnFamilyDefinition = new AquilesColumnFamily
                    {
                        Keyspace = _keySpace,
                        Name = _columnFamily,
                        Comparator = "UTF8Type",
                        Type = AquilesColumnFamilyType.Standard
                    }
                };

                using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName, _keySpace))
                {
                    connection.Execute(cmd);
                }

                RetrieveKeyspacesCommand retrieveKeyspacesCmd = new RetrieveKeyspacesCommand();
                using (IAquilesConnection connection = AquilesHelper.RetrieveConnection(ClusterName))
                {
                    connection.Execute(retrieveKeyspacesCmd);
                }

                keySpace = retrieveKeyspacesCmd.Keyspaces.FirstOrDefault(ks => ks.Name.CompareTo(_keySpace) == 0);

                if (keySpace != default(AquilesKeyspace))
                {
                    if (keySpace.ColumnFamilies.ContainsKey(_columnFamily))
                        LoggerWrapper(MajorEnum.Trace, MinorEnum.ColumnFamilyCreated,
                            "Created ColumnFmily {0} in KeySpace {1}", new object[] { _columnFamily, _keySpace }, logger);
                    else
                    {
                        LoggerWrapper(MajorEnum.Error, MinorEnum.ColumnFamilyCreateFailed,
                            "Could not retrieve ColumnFamily {0} from KeySpace {1} after create", new object[] { _columnFamily, _keySpace }, logger);
                        return;
                    }
                }
                else
                {
                    LoggerWrapper(MajorEnum.Error, MinorEnum.CassandraKeySpaceDoesntExist,
                        "Cannot access KeySpace {0} after create of ColumnFamily {1}", new object[] { _keySpace, _columnFamily }, logger);
                    return;
                }

            }
            catch (Exception ex)
            {
                LoggerWrapper(MajorEnum.Error, MinorEnum.ExceptionCreatingColumnFamily,
                    "Exception creating ColumnFamily {0} in KeySpace {1}: {2}", new object[] { _columnFamily, _keySpace, ex }, logger);
            }
            return;
        }

        #endregion
    }
}
